﻿using System;
using System.Buffers;
using System.Runtime.InteropServices;
using System.Text;
using Deer.Constant;
using Deer.Hash;
using Letter.IO.Binary;

namespace Telex;

sealed partial class RpcChannel
{
    private void OnRcvPipelineMessage(ref SocketPak pak)
    {
        var command = pak.command;
        if (command == Command.SEND || command == Command.REQUEST || command == Command.RESPONSE)
        {
            var hash = HashGenerator.Create(pak.buf.AsSpan());
            var correct = pak.hash == hash;
            if (correct)
            {
                WriteRcvChannel(ref pak);
            }
            this.TryWriteVerifyPack(pak.hash, correct ? (byte)1 : (byte)0);
        }
        else if(command == Command.VERIFY)
        {
            WriteRcvChannel(ref pak);
        }
        else if(command == Command.HEARTBEAT)
        {
            pak.Dispose();
            // Console.WriteLine("收到心跳！~！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！");
        }
    }
    
    private void WriteRcvChannel(ref SocketPak pak)
    {
        if (!this.RcvChannelWriter.TryWrite(pak))
        {
            this.logger.Error("Description Failed to write to thereceiving  channel. command:" + pak.command);
            return;
        }
            
        this.RcvChannelWriter.Flush();
    }
    
    private void OnMessageChannelNotify(object state)
    {
        while(this.RcvChannelReader.TryRead(out var pack))
        {
            System.Threading.ThreadPool.QueueUserWorkItem((o) =>
            {
                var pak = (SocketPak) o;
                var command = pak.command;
                switch (command)
                {
                    case Command.SEND:
                    case Command.REQUEST:
                    case Command.RESPONSE:
                        this.OnRcvMessagePack(command, ref pak);
                        break;
                    case Command.VERIFY:
                        this.OnRcvVerifyPack(command, ref pak);
                        break;
                }

                pak.Dispose();
            },
            pack);
        }

        this.RcvChannelReader.AsyncRcv();
    }
    
    private void OnRcvMessagePack(byte command, ref SocketPak pack)
    {
        var buf = new ReadOnlySequence<byte>(pack.buf.AsMemory());
        var reader = buf.ToBytesReader();

        var mPack = new MessagePack();
        var uidBuf = reader.ReadBuffer(UID_SIZE);
        var span = (Span<byte>)stackalloc byte[UID_SIZE];
        uidBuf.CopyTo(span);
        mPack.uid = MemoryMarshal.Read<Guid>(span);
        mPack.protocolId = reader.ReadUInt32BE();
        mPack.opcode = reader.ReadInt32BE();
        mPack.data = reader.ReadBuffer((int)reader.ReadableLength).ToArray();

        this.MessageRoute(pack.command, ref mPack);
    }
    
    private void OnRcvVerifyPack(byte command, ref SocketPak pack)
    {
        var buf = new ReadOnlySequence<byte>(pack.buf.AsMemory());
        var reader = buf.ToBytesReader();

        var vPack = new VerifyPack();
        vPack.hash = reader.ReadUInt32BE();
        vPack.status = reader.ReadUInt8();

        if (vPack.status == Booleans.TRUE)
        {
            if (this.waitVerifyUnits.TryRemove(vPack.hash, out var unit))
            {
                unit.pak.Dispose();
            }
        }
        else if (vPack.status == Booleans.FALSE)
        {
            if (this.waitVerifyUnits.TryGetValue(vPack.hash, out var unit))
            {
                if (!this.TryWriteChannel(ref unit.pak))
                {
                    unit.pak.Dispose();
                    this.waitVerifyUnits.TryRemove(vPack.hash, out _);
                }
            }
        }
    }

    private void MessageRoute(byte command, ref MessagePack pack)
    {
        switch (command)
        {
            case Command.SEND:
                if (!this.register.TryGetSendProtocolFunc(pack.protocolId, out var sndUnit))
                {
                    this.logger.Error($"The specified protocol id:'{pack.protocolId}' could not be found in the list");
                    return;
                }
                
                sndUnit.Invoke(this.remoteHost, pack.protocolId, pack.data);
                break;
            case Command.REQUEST:
                if (!this.register.TryGetCallProtocolFunc(pack.protocolId, out var callUnit))
                {
                    this.logger.Error($"The specified protocol id:'{pack.protocolId}' could not be found in the list");
                    return;
                }
                
                var uid = pack.uid;
                var protocolId = pack.protocolId;
                callUnit.Invoke(this.remoteHost, pack.protocolId, pack.data, (data, opcode) =>
                {
                    if (!this.TryWriteResponsePak(uid, protocolId, data, opcode))
                    {
                        this.logger.Error($"Failed to write a bidirectional Rpc response message. Procedure Protocol:{protocolId}");
                    }
                });
                break;
            case Command.RESPONSE:
                if (!this.responseUnits.TryRemove(pack.uid, out var unit))
                {
                    this.logger.Error($"The specified id：'{pack.uid}' was not found in the response list");
                    return;
                }
                unit.func.Invoke(pack.data, pack.opcode);
                break;
        }
    }
}